package com.gitee.ixtf.rsocket.verticle

import com.gitee.ixtf.rsocket.IBrokerServiceAgent
import com.gitee.ixtf.rsocket.kotlinx.cloudEvent
import com.gitee.ixtf.rsocket.service.PONG
import com.gitee.ixtf.rsocket.service.PingRequest
import com.gitee.ixtf.vertx.verticle.BaseCoroutineVerticle
import io.cloudevents.CloudEvent
import io.rsocket.Payload
import io.rsocket.RSocket
import io.rsocket.util.DefaultPayload
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactor.asFlux
import kotlinx.coroutines.reactor.flux
import kotlinx.coroutines.reactor.mono
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

abstract class BrokerServiceVerticle : BaseCoroutineVerticle() {
  protected abstract val serviceAgent: IBrokerServiceAgent

  override suspend fun start() {
    super.start()
    serviceAgent.serviceRSocket(
        object : RSocket {
          override fun metadataPush(payload: Payload): Mono<Void> =
              mono { metadataPush(payload.cloudEvent()) }.then()

          override fun fireAndForget(payload: Payload): Mono<Void> =
              mono { fireAndForget(payload.cloudEvent()) }.then()

          override fun requestResponse(payload: Payload): Mono<Payload> = mono {
            val event = payload.cloudEvent()
            when (event.type) {
              PingRequest.CLOUD_EVENT_TYPE -> DefaultPayload.create(PONG)
              else -> requestResponse(event)
            }
          }

          override fun requestStream(payload: Payload): Flux<Payload> = flux {
            requestStream(payload.cloudEvent()).collect { send(it) }
          }

          override fun requestChannel(payloads: Publisher<Payload>): Flux<Payload> =
              mono { requestChannel(payloads.asFlow().map { it.cloudEvent() }) }
                  .flatMapMany { it.asFlux() }
        })
    serviceAgent.rSocketClient()
  }

  override suspend fun stop() {
    serviceAgent.dispose()
    super.stop()
  }

  protected open suspend fun metadataPush(event: CloudEvent) = Unit

  protected open suspend fun fireAndForget(event: CloudEvent) = Unit

  protected abstract suspend fun requestResponse(event: CloudEvent): Payload

  protected open suspend fun requestStream(event: CloudEvent): Flow<Payload> = TODO()

  protected open suspend fun requestChannel(events: Flow<CloudEvent>): Flow<Payload> = TODO()
}
